package com.shujia.jinjie;

import com.shujia.util.HBaseTool;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.PrefixFilter;
import org.apache.hadoop.hbase.filter.QualifierFilter;
import org.apache.hadoop.hbase.util.Bytes;
import redis.clients.jedis.Jedis;

import java.util.Scanner;

/*
    通过redis构建二级索引
 */
public class HBaseIndexWithRedisDemo {
    private static Jedis jedis = new Jedis("master", 7000);
    private static Connection conn = HBaseTool.HBASE_CONN;
    private static Admin admin = HBaseTool.HBASE_ADMIN;


    public static void main(String[] args) throws Exception {
        //创建二级索引表
//        createIndexInRedis();

        Scanner sc = new Scanner(System.in);
        System.out.println("请输入要查询的学生姓名：");
        String name = sc.next();
        selectClazzWithName(name);

    }

    public static void selectClazzWithName(String name) throws Exception {
        //先去redis中查询该学生对应的学号
        String id = jedis.hget("students_index1", name);

        //去hbase中查询
        //因为id在hbase中的students里是作为行键的存在，可以直接使用专用过滤器
        PrefixFilter prefixFilter = new PrefixFilter(Bytes.toBytes(id));

        Get get = new Get(Bytes.toBytes(id));
        Table students = conn.getTable(TableName.valueOf("students"));
        Result result = students.get(get);

        String clazz = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("clazz")));


        System.out.println("学生：" + name + " 所在班级：" + clazz);


    }

    public static void createIndexInRedis() throws Exception {
        //将姓名作为redis的键，学号作为值存储在redis
        //先查询原表数据
        Table students = conn.getTable(TableName.valueOf("students"));

        //使用列名过滤器，得到所有的name单元格
        QualifierFilter qualifierFilter = new QualifierFilter(CompareOperator.EQUAL, new BinaryComparator(Bytes.toBytes("name")));
        Scan scan = new Scan();
        scan.setFilter(qualifierFilter);

        ResultScanner resultScanner = students.getScanner(scan);

        for (Result result : resultScanner) {
            String id = Bytes.toString(result.getRow());
            String name = Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name")));
            jedis.hset("students_index1", name, id);
        }

    }
}
